package com.example.completionservice.aatest2;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


/** 对CompletionService封装，实现快速处理业务任务并汇总结果 https://blog.csdn.net/zhaozhenzuo/article/details/52138140
 *
 * 利用await及notify实现的CompletionService封装类，提高处理速度<br/>
 *
 * 例:<br/>
 *
 * <pre>
 * 1.定义执行业务任务浅程池
 * ExecutorService executorService = Executors.newFixedThreadPool(50);
 *
 * 2.创建快速处理器,传入执行业务线程池，及汇总业务处理结果处理类
 * final QuickProcessByNotify<String, Long>
 * quickProcess = new QuickProcessByNotify<String, Long>(executorService, new QuickTaskResultHandler<String, Long>() {<br/>
 * <em>//业务结果汇总处理类,这里最终汇总结果是long类型的成功数successCount<br/>
 * private AtomicLong successCount = new AtomicLong(0);
 *
 *     public void handle(Future<String> future) { //这里放业务汇总逻辑 }
 *
 *     public Long getResult() { //返回业务汇总结果 return successCount.get(); }
 *           });
 *
 * 3.调用processByCallBack方法，开始查询业务数据并处理
 *
 *           long allCount = quickProcess.processByCallBack(new
 *           QuickBizProcessCallbak() {
 *           //QuickBizProcessCallbak是业务处理回调接口，业务方实现此接口
 *           public void doBiz() { quickProcess.submit(new Callable<String>() {
 *           //一定要提交业务任务到快速处理器 public String call() throws Exception {
 *           return "success"; } }); } });
 *
 * <pre/>
 *
 * @author shenxiu
 *
 * @param <T>
 * @param <R>
 */

public class QuickProcessByNotify<T/* 业务任务处理结果类型 */, R/* 最终汇总的结果类型 */> implements QuickProcessService<T, R> {

    private static final Logger logger = LoggerFactory.getLogger(QuickProcessByNotify.class);

    /**
     * 发送者发送完标记
     */
    private volatile boolean sendFinishFlag;

    private CompletionService<T> quickCompletionService;

    private Lock lock = new ReentrantLock();

    private Condition notEmptyCondition = lock.newCondition();

    private AtomicLong countShouldProcess = new AtomicLong(0);

    /**
     * 当前处理器状态<br/>
     */
    private volatile int state;

    private static int RUNNING_STATE = 0;

    private static int TERMINATE_STATE = 1;

    /**
     * 消费者线程
     */
    private Thread consumeThread;

    /**
     * 业务方实现结果处理类
     */
    private QuickTaskResultHandler<T, R> taskResultHandler;

    public QuickProcessByNotify(CompletionService<T> quickCompletionService,
                                QuickTaskResultHandler<T, R> taskResultHandler) {
        this.quickCompletionService = quickCompletionService;
        this.taskResultHandler = taskResultHandler;
    }

    public QuickProcessByNotify(ExecutorService executorService,
                                QuickTaskResultHandler<T, R> taskResultHandler) {
        quickCompletionService = new ExecutorCompletionService<T>(executorService);
        this.taskResultHandler = taskResultHandler;
    }

    public void submit(Callable<T> task) {
        //把结果集放入 quickCompletionService
        quickCompletionService.submit(task);

        long cnt = countShouldProcess.incrementAndGet();

        if (cnt == 1) {
            this.notifyConsumer();
        }

    }



    /**
     * 消费者消费任务结果runable
     *
     * @author shenxiu
     *
     */
    public class DefaultConsumeTask implements Runnable {

        public void run() {

            long count;
            while (true) {
                /**
                 * 1.判断退出条件<br/>
                 * 若sendFinishFlag为true,
                 * 则发送方之前在做submit的时候肯定会先增加countShouldProcess值<br/>
                 * 也就是countShouldProcess的值变化先行发生于sendFinishFlag值的变化<br/>
                 * 因此不会发生sendFinishFlag为true，但消费者看到的countShouldProcess为0，
                 * 实际有消息未处理的情况
                 */
                if (sendFinishFlag && countShouldProcess.get() <= 0) {
                    /**
                     * 发送方已经发送完数据，且消费队列没有数据了
                     */
                    break;
                }
                lock.lock();
                try {
                    /**
                     * 2. 发送方未发送完数据，但消费队列还没有数据，await下，这里加锁防止以下情况：<br/>
                     * 消费者在判断sendFinishFlag的条件，看到的是false，
                     * 然后也看到countShouldProcess值为0<br/>
                     * 消费者在要做await时，未加锁的情况下生产者将sendFinishFlag设置为true了，
                     * 这时消费者接着await，然后消费者就永远处理await了。
                     */
                    if (!sendFinishFlag && countShouldProcess.get() <= 0) {
                        notEmptyCondition.await();
                    }
                } catch (InterruptedException e) {
                    logger.error("quickProcessByNotify consumer thread,InterruptedException", e);
                } finally {
                    lock.unlock();
                }
                /**
                 * 3.有数据，取出处理
                 */
                count = countShouldProcess.get();
                logger.info("--->count: " + count);
                for (int i = 0; i < count; i++) {
                    try {
                        Future<T> future = quickCompletionService.take();
                        taskResultHandler.handle(future);
                    } catch (InterruptedException e) {
                        logger.error("InterruptedException err");
                    } finally {
                        countShouldProcess.decrementAndGet();
                    }
                }

            }

            // 消费者正常退出了，设置state为TERMIATE
            state = TERMINATE_STATE;
        }
    }

    public void waitProcessResult() throws InterruptedException {
        consumeThread.join();
    }

    public void notifyConsumer() {
        try {
            lock.lock();
            notEmptyCondition.signal();
             logger.info("===sig");
        } finally {
            lock.unlock();
        }
    }

    public void setSendFinishFlagToTrue() {
        try {
            lock.lock();
            sendFinishFlag = true;
            notEmptyCondition.signal();
        } finally {
            lock.unlock();
        }
    }

    public void shutdownNow() {
        try {
            lock.lock();

            if (state == TERMINATE_STATE) {
                //当前消费者已经退出
                return;
            }

            //否则强制退出
            logger.error("err consume thread not stop,shutdownNow");
            sendFinishFlag = true;
            countShouldProcess.addAndGet(-1);
            notEmptyCondition.signal();
            consumeThread.interrupt();
        } catch (Exception e) {
            logger.error("shutdown err", e);
        } finally {
            lock.unlock();
        }
    }

    public void setTaskResultHandler(QuickTaskResultHandler<T, R> taskResultHandler) {
        this.taskResultHandler = taskResultHandler;
    }

    public Long getResult() {
        return taskResultHandler.getResult();
    }

    public void start() {
//        Assert.assertNotNull(quickCompletionService);
//        Assert.assertNotNull(taskResultHandler);

        consumeThread = new Thread(new DefaultConsumeTask());
        consumeThread.start();
        state = RUNNING_STATE;
    }

    public Long processByCallBack(QuickBizProcessCallbak bizProcessCallbak) {
        try {
            // 1.开启处理器
            this.start();

            // 2.执行业务逻辑
            bizProcessCallbak.doBiz();

            logger.info("before:" + getResult());

            // 3.业务数据都发送完了，设置sendFinishFlag为true
            this.setSendFinishFlagToTrue();

            // 3.等待consumeThread消费
            consumeThread.join();

            logger.info("after:" + getResult());

            // 4.返回结果
            return this.getResult();
        } catch (Exception e) {
            logger.error(">processByCallBack err", e);
        } finally {
            // 非正常情况下关闭消费者线程
            this.shutdownNow();
        }

        return this.getResult();
    }

}
